package gg.zsw.datasource.demo.kafka;

import gg.zsw.dataframework.decode.DecodeChain;
import gg.zsw.dataframework.pubsub.AbsSequentialSubject;
import gg.zsw.dataframework.pubsub.AbstractObserver;

/**
 * 作为观察者，收听MessageTopic分类出来的消息。
 * 分布式系统中，kafka消息并不是进程都要使用的，用不到的数据就不要注册。
 * <p>
 * 作为主题，发布程序能够直接使用的数据类型。
 * 数据类型的转换，通过DecodeChain来实现。
 * 有时多个子模块会关注同一数据，但所需类型各不相同，这时采用不同的DecodeChain即可。
 * 没有约束具体发布的数据类型，也没有约束具体的观察者，保持灵活性
 *
 * @author zhou.sw
 * @Date 2020/6/30 11:24
 */
public abstract class MessageSubject<Observer extends AbstractObserver<NotifyData>, NotifyData>
        extends AbsSequentialSubject<Observer, NotifyData>
        implements AbstractObserver<String> {

    /**
     * 处理链，很多场景下，是不需要复杂的处理链的，只需要1个
     */
    private DecodeChain<String, NotifyData> decodeChain;

    public MessageSubject(DecodeChain<String, NotifyData> decodeChain) {
        this.decodeChain = decodeChain;
    }


    @Override
    public void update(String msg) {
        // 处理可能是比较费劲的，如果无人关注这个主题，那就不需要处理
        if (observers.isEmpty()) {
            return;
        }
        // 数据链转换后，全量通知
        NotifyData message = decodeChain.handle(msg);
        notifyAll(message);
    }

    @Override
    protected void notify(NotifyData notifyData) {

    }

    @Override
    public void nonObserverCallBack() {

    }
}
